package install.ren.exercise.test.zookeeper;

import com.google.common.base.Charsets;
import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;

import java.security.NoSuchAlgorithmException;
import java.util.List;

/**
 * @author chenyuwei
 * @date 2018/12/11
 * zookeeper数据复制工具
 * 启动参数1为来源zk地址
 * 启动参数2为目标zk地址
 * 启动参数3为需要复制的路径，此路径下的子节点都会被复制
 */
public class ZookeeperCopy {

    /**
     * 来源zk客户端
     */
    private static CuratorFramework sourceZK;

    /**
     * 目标zk客户端
     */
    private static CuratorFramework targetZK;

    /**
     * 来源zk地址
     */
    private static String sourceZKAddress;

    /**
     * 目标zk地址
     */
    private static String targetZKAddress;

    private static String SCHEME = "digest";
    private static String redisAdminPassword = "redis_admin:redis_admin";
    private static String redisRWPassword = "redis_rw:redis_rw";

    public static void main(String[] args) throws Exception {
        sourceZKAddress = ":2181";
        targetZKAddress = "?:2181";
        boolean acl = true;


        if (sourceZK == null) {
            sourceZK = getSourceZK(acl, redisAdminPassword);
            System.out.println("source connect success");
        }
        if (targetZK == null) {
            targetZK = getTargetZK(acl, redisAdminPassword);
            System.out.println("target connect success");
        }
        //                "redis_token_info","redis_info"
        copyPath(
                "redis_token_info");

    }

    public static void copyPath(String... paths) throws Exception {

        for (String nodePath : paths) {
            nodePath = "/" + nodePath;
            //目标复制路径
            //String nodePath = "/jdbc_namespace";
            System.out.println("nodePath:" + nodePath);
            final Stat statSource = sourceZK.checkExists().forPath(nodePath);
            final Stat statTarget = targetZK.checkExists().forPath(nodePath);
            if (statSource == null) {
                System.out.printf("退出：source 节点不存在:%s", nodePath);
                //临时节点不创建
                return;
            }
            if (statSource.getEphemeralOwner() > 0) {
                System.out.printf("退出：source 临时节点:%s", nodePath);
                return;
            }

            if (statTarget != null) {
                System.out.printf("退出：target 节点已存在:%s", nodePath);
                //临时节点不创建
                return;
            }

            // 根节点创建
            byte[] data = sourceZK.getData().forPath(nodePath);
            targetZK.create().withACL(ZooDefs.Ids.CREATOR_ALL_ACL).forPath(nodePath, data);

            System.out.println("数据开始复制" + nodePath);
            getNodes(nodePath);
            System.out.println("数据复制完成" + nodePath);
        }
    }

    private static CuratorFramework getCuratorFramework(String url, String password, List<ACL> aclList) {

        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
        //默认情况使用json结构来操作zk的data，所以使用 {} 作为默认值
        builder.defaultData("{}".getBytes(Charsets.UTF_8))
                .connectString(url)
                .retryPolicy(new ExponentialBackoffRetry(20000, 3))
                .sessionTimeoutMs(30000)
                .connectionTimeoutMs(30000);
        if (password != null) {
            builder.authorization(SCHEME, password.getBytes());
        }
        if (aclList != null) {
            builder.aclProvider(new ACLProvider() {
                @Override
                public List<ACL> getDefaultAcl() {
                    return null;
                }

                @Override
                public List<ACL> getAclForPath(String path) {
                    return null;
                }
            });
        }
        return builder.build();
    }


    private static CuratorFramework getTargetZK(boolean acl, String password) throws Exception {
        if (targetZK == null) {
            if (acl) {
                targetZK = getCuratorFramework(targetZKAddress, password, getACLList());
            } else {
                targetZK = CuratorFrameworkFactory.newClient(targetZKAddress, new ExponentialBackoffRetry(20000, 3));
            }
            targetZK.start();
        }
        return targetZK;
    }

    private static CuratorFramework getSourceZK(boolean acl, String password) throws Exception {

        if (sourceZK == null) {
            if (acl) {
                sourceZK = getCuratorFramework(sourceZKAddress, password, getACLList());
            } else {
                sourceZK = CuratorFrameworkFactory.newClient(sourceZKAddress, new ExponentialBackoffRetry(20000, 3));
            }
            sourceZK.start();
        }
        return sourceZK;
    }

    private static void getNodes(String nodePath) throws Exception {
        List<String> list = sourceZK.getChildren().forPath(nodePath);
        final long ephemeralOwner = sourceZK.checkExists().forPath(nodePath).getEphemeralOwner();
        if (ephemeralOwner > 0) {
            //临时节点不创建
            return;
        }
        for (String str : list) {
            String subNodePath = nodePath + "/" + str;
            Stat exists = sourceZK.checkExists().forPath(subNodePath);
            if (exists.getEphemeralOwner() > 0) {
                //临时节点不创建
                System.out.println("临时节点不创建" + subNodePath);
                continue;
            } else {
                if (targetZK.checkExists().forPath(subNodePath) != null) {
                    System.out.println("节点已存在不复制" + subNodePath);
                    continue;
                }
                System.out.println("正在创建" + subNodePath);
                byte[] data = sourceZK.getData().forPath(subNodePath);
                targetZK.create().withACL(ZooDefs.Ids.CREATOR_ALL_ACL).forPath(subNodePath, data);
            }
            getNodes(subNodePath);
        }

    }
    private static List<ACL> getACLList() throws NoSuchAlgorithmException {
        Id admin = new Id(SCHEME, DigestAuthenticationProvider.generateDigest(redisAdminPassword));
        Id redisRW = new Id(SCHEME, DigestAuthenticationProvider.generateDigest(redisRWPassword));
        ACL aclAdmin = new ACL(ZooDefs.Perms.ALL, admin);
        ACL aclWrite = new ACL(ZooDefs.Perms.WRITE, redisRW);
        ACL aclRead = new ACL(ZooDefs.Perms.READ, redisRW);
        return Lists.newArrayList(aclAdmin, aclWrite, aclRead);
    }
}
